package com.central.scheduler.worker.registry;

import com.central.scheduler.service.registry.HeartBeatTask;
import com.central.scheduler.service.registry.RegistryClient;
import com.central.scheduler.common.Constants;
import com.central.scheduler.common.IStoppable;
import com.central.scheduler.common.enums.NodeType;
import com.central.scheduler.common.utils.NetUtils;
import com.central.scheduler.remote.utils.NamedThreadFactory;
import com.central.scheduler.worker.config.WorkerConfig;
import com.central.scheduler.worker.runner.WorkerManagerThread;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static com.central.scheduler.common.Constants.*;

/**
 * @author Tindy
 * @date 2021/10/29
 * @describe
 */
@Service
public class WorkerRegistryClient {

    private final Logger logger = LoggerFactory.getLogger(WorkerRegistryClient.class);
    /**
     * worker config
     */
    @Autowired
    private WorkerConfig workerConfig;

    private RegistryClient registryClient;
    /**
     * worker manager
     */
    @Autowired
    private WorkerManagerThread workerManagerThread;
    /**
     * heartbeat executor
     */
    private ScheduledExecutorService heartBeatExecutor;
    /**
     * worker startup time, ms
     */
    private long startupTime;
    @PostConstruct
    public void initWorkRegistry() {
        this.startupTime = System.currentTimeMillis();
        this.registryClient = RegistryClient.getInstance();
        this.heartBeatExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("HeartBeatExecutor"));
    }

    /**
     * registry
     */
    public void registry() {
        String address = NetUtils.getAddr(workerConfig.getListenPort());
        String workerZkPath = getWorkerZkPath();
        int workerHeartbeatInterval = workerConfig.getWorkerHeartbeatInterval();

        registryClient.persistEphemeral(workerZkPath, "");
        logger.info("worker node : {} registry to ZK {} successfully", address, workerZkPath);

        HeartBeatTask heartBeatTask = new HeartBeatTask(startupTime,
                workerConfig.getWorkerMaxCpuloadAvg(),
                workerConfig.getWorkerReservedMemory(),
                workerConfig.getHostWeight(),
                workerZkPath,
                Constants.WORKER_TYPE,
                registryClient,
                workerConfig.getWorkerExecThreads(),
                workerManagerThread.getThreadPoolQueueSize()
        );

        this.heartBeatExecutor.scheduleAtFixedRate(heartBeatTask, workerHeartbeatInterval, workerHeartbeatInterval, TimeUnit.SECONDS);
        logger.info("worker node : {} heartbeat interval {} s", address, workerHeartbeatInterval);
    }
    /**
     * get worker path
     */
    public String getWorkerZkPath() {
        String address = getLocalAddress();
        String path=REGISTRY_SCHEDULER_WORKERS+SINGLE_SLASH+address;
        return path;
    }
    /**
     * get local address
     */
    private String getLocalAddress() {
        return NetUtils.getAddr(workerConfig.getListenPort());
    }
    public void setRegistryStoppable(IStoppable stoppable) {
        registryClient.setStoppable(stoppable);
    }

    public void handleDeadServer(String node, NodeType nodeType, String opType) throws Exception {
        registryClient.handleDeadServer(node, nodeType, opType);
    }
    /**
     * remove registry info
     */
    public void unRegistry() {
        String address = getLocalAddress();
        String workerZkPath = getWorkerZkPath();
        registryClient.remove(workerZkPath);
        logger.info("worker node : {} unRegistry from ZK {}.", address, workerZkPath);
        this.heartBeatExecutor.shutdownNow();
        logger.info("heartbeat executor shutdown");
        registryClient.close();
    }
}
